package org.hadoop.dao;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.hadoop.entity.TitleHot;
import org.springframework.stereotype.Repository;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

@Repository
public class HBseDao {
    //获取连接
    public Connection getConnection() throws IOException {
        //设置本机客户端访问路径
        System.setProperty("hadoop.home.dir","E:\\hadoop");
        //设置访问身份
        System.setProperty("HADOOP_USER_NAME","root");
        //获取配置对象
        Configuration config = HBaseConfiguration.create();
        //hbase客户端配置zookeeper的集群地址
        config.set("hbase.zookeeper.quorum", "master,slave1,slave2");
        config.set("hbase.zookeeper.property.clientPort", "2181");
        return ConnectionFactory.createConnection(config);
    }

    /**
     * 根据时间段去查询
     * @param start
     * @param end
     * @return
     * @throws Exception
     */
    public List<TitleHot> query(long start, long end) throws Exception {
        Connection connection = this.getConnection();
        //获取table对象 操作table
        Table table = connection.getTable(TableName.valueOf("title_hot"));
        List<TitleHot> list = new ArrayList<>();
        //获取scan对象
        Scan scan = new Scan();
        //得到某一个时间段的数据
        scan.setTimeRange(start,end);
        //获取全部数据
        //遍历
        //行数据
        for (Result next : table.getScanner(scan)) {
            //获取下一行数据
            //获取行键
            byte[] row = next.getRow();
            //单元格数据集合
            List<Cell> cells = next.listCells();
            TitleHot titleHot = new TitleHot();
            for (Cell cell : cells) {//cell:每个单元格
                //装载id——> rowKey
                titleHot.setId(Bytes.toString(CellUtil.cloneRow(cell)));
                //获取列和值
                String col = Bytes.toString(CellUtil.cloneQualifier(cell));
                String value = Bytes.toString(CellUtil.cloneValue(cell));
                switch (col) {
                    case "userName":
                        titleHot.setUsername(value);
                        break;
                    case "agentType":
                        titleHot.setAgentType(value);
                        break;
                    case "title":
                        titleHot.setTitle(value);
                        break;
                    case "time":
                        titleHot.setTime(value);
                        break;
                }
            }
            //装载行信息
            list.add(titleHot);

        }
        //关闭
        table.close();
        connection.close();
        return list;

    }
}
